package com.item.spark;

import cn.hutool.json.JSONUtil;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

import java.util.Arrays;
import java.util.List;

public class sparkCoreDemo {
    public static void main(String[] args) {

        // 0. TODO 准备环境
        SparkConf sparkConf = new SparkConf().setAppName("JavaSparkDemo").setMaster("local[*]");
        JavaSparkContext jsc = new JavaSparkContext(sparkConf);
        jsc.setLogLevel("WARN");

        //1, TODO 准备数据
        JavaRDD<String> fileRDD = jsc.textFile("data/input/words.txt");

        //2, TODO 处理数据 -WordCount
        //切割

        // java的函数 /lambda表达式  jdk8新语法
        JavaRDD<String> wordRDD = fileRDD.flatMap(line-> Arrays.asList(line.split(" ")).iterator());

        //每个单词记为1
        JavaPairRDD<String,Integer> wordAndOneRDD = wordRDD.mapToPair(word-> new Tuple2<>(word,1));
        //分组聚合
        JavaPairRDD<String,Integer> wordAndCountRDD = wordAndOneRDD.reduceByKey((a,b)-> a + b);

        //. TODO 输出结果
        List<Tuple2<String,Integer>> result = wordAndCountRDD.collect();

        //原始写法输出
        result.forEach(t-> System.out.println(t));
        //表达式简写
        result.forEach(System.out::println);

        //4.TODO 关闭资源
        jsc.stop();

    }
}
